package rabbit.releaseAck;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import rabbit.util.RabbitMqUtils;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 * ClassName: ConfirmMessage
 * Package: rabbit.releaseAck
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-10-25
 * Time: 18:21
 */

//发布确认模式
//1.单个确认  比较时间 观察效率最高的
//2.批量确认
//3.异步批量确认

public class ConfirmMessage {

    //批量发送消息
    public static final int message_count = 1000;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {

        //单个确认
        //publishMessageOne();
        //1073ms

        //批量确认 但是弊端不知道哪些消息没有被确认到
        //publishMessageBatch();
        //92ms

        publishMessageAsync();
        //37ms
    }

    public static void publishMessageAsync() throws IOException, TimeoutException {
        //获取信道
        Channel chanel = RabbitMqUtils.getChanel();
        //队列名称
        String QUEUE_NAME = UUID.randomUUID().toString();
        //使用信道 声明队列
        chanel.queueDeclare(QUEUE_NAME, true, false, false, null);

        //开启发布确认模式
        chanel.confirmSelect();

        //线程安全有序的哈希表 使用高并发
        //1.轻松的将序号和消息进行管理
        //2.轻松的批量删除
        //3.支持高并发
        ConcurrentSkipListMap<Long, String> outConfirms = new ConcurrentSkipListMap<>();

        //准备监听器 监听哪些消息成功了 哪些失败了
        //两个监听的接口 一个是成功的 一个是失败的
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {

            if (multiple) {
                //如果是批量
                ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap =
                        outConfirms.headMap(deliveryTag);
            } else {
                outConfirms.remove(deliveryTag);
            }
            //消息接收成功的回调函数
            System.out.println("确定的消息: " + deliveryTag);

        };

        /**
         * 1.消息的标记
         * 2.是否未批量却仍
         */
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            //消息接收失败的回调函数
            //记录下来 在重发 后续在处理
            String s = outConfirms.get(deliveryTag);
            System.out.println("内容是: " + s + "未确定的消息编号: " + deliveryTag);
        };

        //成功 失败
        chanel.addConfirmListener(ackCallback, nackCallback);
        //异步监听

        //记录开始时间
        long begin = System.currentTimeMillis();
        for (int i = 0; i < message_count; i++) {
            String message = "消息: " + i;
            chanel.basicPublish("", QUEUE_NAME, null, message.getBytes());

            //异步的 传统方式就在这个却仍 异步是broker通知你
            //所以要有一个监听器

            //处理未确认消息 使用并发链路队列 ConcurrentLinkedQueue
            //序号在信道里面
            outConfirms.put(chanel.getNextPublishSeqNo(), message);

            //1.是由一个队列 将发送的全部保存起来
            //2.去确认的那里 删除已经发送的消息

        }

        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("异步" + message_count + "异步确认消息,耗时: " + (end - begin) + "ms");


    }

    //批量确定
    public static void publishMessageBatch() throws IOException, InterruptedException, TimeoutException {
        //获取信道
        Channel chanel = RabbitMqUtils.getChanel();
        //队列名称
        String QUEUE_NAME = UUID.randomUUID().toString();
        //使用信道 声明队列
        chanel.queueDeclare(QUEUE_NAME, true, false, false, null);

        //开启发布确认模式
        chanel.confirmSelect();
        //记录开始时间
        long begin = System.currentTimeMillis();

        //批量确认消息的大小
        int batchSize = 100;

        //发送消息
        for (int i = 0; i < message_count; i++) {
            String message = i + "";
            chanel.basicPublish("", QUEUE_NAME, null, message.getBytes());

            //达到100确认
            if (i % batchSize == 0) {
                //批量发布确认
                boolean b = chanel.waitForConfirms();
            }
        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("批量" + message_count + "批量确认消息,耗时: " + (end - begin) + "ms");


    }


    //单个确定
    public static void publishMessageOne() throws IOException, TimeoutException, InterruptedException {
        //获取信道
        Channel chanel = RabbitMqUtils.getChanel();
        //队列名称
        String QUEUE_NAME = UUID.randomUUID().toString();
        //使用信道 声明队列
        chanel.queueDeclare(QUEUE_NAME, true, false, false, null);

        //开启发布确认模式
        chanel.confirmSelect();
        //记录开始时间
        long begin = System.currentTimeMillis();

        //发送消息
        for (int i = 0; i < message_count; i++) {
            String message = i + "";
            chanel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            //单个消息马上就进行发布确认
            boolean b = chanel.waitForConfirms();
        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("发布" + message_count + "单个确认消息,耗时: " + (end - begin) + "ms");

    }
}
